package com.learn.mq.listener;

import com.alibaba.fastjson2.JSON;
import com.learn.mq.beans.Person;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.UnsupportedEncodingException;
import java.util.List;

/**
 * ClassName:Listener1Msg
 * Package:com.learn.mq.listener
 * Description:
 *
 * @Author:@郭先森
 * @Create:2025/7/30 - 7:34
 * @Version:v1.0
 */
@Component
@Slf4j
public class Listener1Msg {
    @Autowired
    @Qualifier("defaultMQPushConsumer")
    private DefaultMQPushConsumer learnClientConsumer;
    @PostConstruct
    public void init(){
        try {
            startConsumer();//初始化的时候就注册这个消费者
        } catch (MQClientException e) {
            throw new RuntimeException(e);
        }
    }
    //消费者启动的方法
    public void startConsumer() throws MQClientException {
        learnClientConsumer.setConsumerGroup("LearnClientConsumer");
        learnClientConsumer.subscribe("GXS-LEARN-CLIENT", "GXS");
        learnClientConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,
                                                            ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for(MessageExt msg : list){
                    String topic = msg.getTopic();
                    String tags = msg.getTags();
                    String keys = msg.getKeys();
                    String body = null;
                    try {
                        body = new String(msg.getBody(), "UTF-8");
                    } catch (UnsupportedEncodingException e) {
                        throw new RuntimeException(e);
                    }
                    log.info("监听器1收到了消息，收到的消息信息为：TOPIC:{},TAGS:{},KEYS:{},BODY:{}",topic,tags,keys,body);
                    processMessage(body);
                }
                //消费成功
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        learnClientConsumer.start(); //启动消费者
    }

    public void processMessage(String body){
        Person person = JSON.parseObject(body, Person.class);
        log.info("监听器1解析出来了消息"+person.toString());
        log.info("执行业务中");
        log.info("执行业务完成");
    }

}
